package com.elegant.code.spi.web.gateway.zk;

import com.elegant.code.spi.web.gateway.Constants;
import com.elegant.code.spi.web.gateway.ServicesHolder;
import com.elegant.code.spi.web.gateway.plugins.divide.LoadBalance;
import lombok.extern.slf4j.Slf4j;
import org.I0Itec.zkclient.ZkClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.web.reactive.context.ReactiveWebServerInitializedEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Objects;

@Slf4j
@Service
public class ServiceRegisterAndDiscoverImpl
        implements ServiceRegisterAndDiscover, ApplicationListener<ReactiveWebServerInitializedEvent> {

    @Autowired
    private ZkClient zkClient;
    private String registryPath = Constants.ZK_REGISTRY;

    @Override
    public void onApplicationEvent(ReactiveWebServerInitializedEvent event) {
        registryRootPath();
        registryServicePath();
    }

    private void registryRootPath() {
        if (!zkClient.exists(registryPath)) {
            zkClient.createPersistent(registryPath);
            log.info(">>> create registry node:" + registryPath);
        }

    }

    private void registryServicePath(){
        ServicesHolder.getServiceNames()
                .stream()
                .map(serviceName -> {
                    String servicePath = registryPath + "/" + serviceName;
                    ServicesHolder.addServicePath(serviceName, servicePath);
                    return servicePath;
                })
                .filter(servicePath -> !zkClient.exists(servicePath))
                .peek(servicePath -> {
                    //创建service节点（持久）
                    zkClient.createPersistent(servicePath);
                    log.info(">>>create service node:" + servicePath);
                });
    }

    public void registryAddressPath(String servicePath, String address){
        String addressPath = servicePath + "/address-";
        ServicesHolder.addAddress(servicePath, address);
        String addressNode = zkClient.createEphemeralSequential(addressPath, address);
        log.info(">>> create address node:" + addressNode);
    }

    public String discover(String serviceName, LoadBalance loadBalance) {
        String servicePath = ServicesHolder.getServicePath(serviceName);

        //获取服务节点
        if (!zkClient.exists(servicePath)) {
            throw new RuntimeException(String.format(">>>can't find any service node on path {}",servicePath));
        }

        //从本地缓存获取某个服务地址

        String address = loadBalance.load(serviceName);
        if (Objects.isNull(address)){
            List<String> addressList = zkClient.getChildren(servicePath);
            ServicesHolder.addAddress(servicePath, addressList);

            //监听servicePath下的子文件是否发生变化
            zkClient.subscribeChildChanges(servicePath,(parentPath,currentChilds)->{
                log.info(">>>servicePath is changed:" + parentPath);
                ServicesHolder.addAddress(servicePath, currentChilds);
            });

            if (CollectionUtils.isEmpty(addressList)) {
                throw new RuntimeException(String.format(">>>can't find any address node on path {}", servicePath));
            }
            address = loadBalance.load(serviceName);
            log.info(">>>get address node:" + address);
        }
        return address;
    }
}
